package com.comtop.db.fastar

import java.io.File
import org.apache.spark.SparkContext
import org.apache.spark.rdd.JdbcRDD
import java.io.PrintWriter
import org.apache.spark.sql.types.StructType
import org.apache.spark.sql.types.StructField
import org.apache.spark.sql.types.StringType
import java.util.Date
import java.text.SimpleDateFormat
import java.sql.Timestamp
import com.comtop.db.utils.dateUtils
import org.apache.ivy.util.DateUtil
import scala.collection.mutable.ArrayBuffer

class FastarUtils {

  case class PointData(name: String, value: Double, time_mark: Date)

  /**
   * 引用winutils工具
   */
  def mkdirFile(): Unit = {
    val workgroud = new File(".");
    System.getProperties().put("hadoop.home.dir",
      workgroud.getAbsoluteFile());
    new File("./bin").mkdir();
    new File("./bin/winutils.exe").createNewFile();
  }

  /**
   * 获取fastar数据
   */
  def getDataFromFastar() = {

    val sc = JdbcConnection.sc
    println("***************************************start***************************************")
    val starTime = System.currentTimeMillis()
    val rdd = new JdbcRDD(
      //      sc, getConnection,"SELECT  NAME, TIME_MARK, VALUE, VALUE_TYPE  FROM DATA_TABLE WHERE NAME='XXX_P_TOTAL_S_00000000'",
      sc, JdbcConnection.getConnection, "SELECT  *  FROM DATA_TABLE WHERE NAME='FOR-TEST-POINT-NAME-00000000'  AND TIME_MARK >= '2016-03-30 14:42:15:000' AND TIME_MARK < '2017-03-31 14:43:21:000'",
      1, 10, 1,
      r => (r.getString("NAME"), r.getDate("TIME_MARK"), r.getString("VALUE")))
    //    rdd.collect().foreach(println)
    //  println(rdd.collect().toList)

    if (JdbcConnection.filePath != "") {
      val writer = new PrintWriter(new File(JdbcConnection.filePath))
      rdd.collect().foreach(x => writer.println(x))
    } else {
      println("[filePath]:获取保存文件路径失败....")
    }

    println("*********************共查询到" + rdd.count() + "条数据...")

    sc.stop()
    println("***************************************end***************************************")
    val endTime = System.currentTimeMillis()
    println("[共耗时]: " + (endTime - starTime) + "毫秒")
  }

  /**
   * 计算某个小时的电量
   * 日期格式：yyyy-MM-dd hh:mm:ss:fff
   */
  def getEQByHours(sql: String, startTime: String, endTime: String): Double = {
    var electricQuantity = 0.0
    val rdd = new JdbcRDD(JdbcConnection.sc, JdbcConnection.getConnection, sql,
      1, 10, 1, r => (r.getString("NAME"), r.getDouble("VALUE"), r.getTimestamp("TIME_MARK")))
    val sTime = dateUtils.getTimestamp(startTime)
    val eTime = dateUtils.getTimestamp(endTime)
    if (sTime != null && eTime != null) {
      val valueArr = rdd.filter(line => line._3.compareTo(sTime) >= 0 && line._3.compareTo(eTime) < 0)
        .map(x => x._2).collect()
      electricQuantity = valueArr.apply(valueArr.length - 1) - valueArr.apply(0)
      println(sTime + " 到 " + eTime + " 共消耗电量为： " + electricQuantity)
    } else {
      val valueArr = rdd.map(x => x._2).collect()
      electricQuantity = valueArr.apply(valueArr.length - 1) - valueArr.apply(0)
      println("共消耗电量为： " + electricQuantity)
    }

    return electricQuantity
  }

  /**
   *  计算某天每个小时的电量
   *  日期格式：yyyy-MM-dd
   */
  def getEQByDays(sql: String, startDay: String, endDay: String): ArrayBuffer[Double] = {
    var hourEQArr = new ArrayBuffer[Double]
    val rdd = new JdbcRDD(JdbcConnection.sc, JdbcConnection.getConnection, sql,
      1, 100, 1, r => (r.getString("NAME"), r.getDouble("VALUE"), r.getTimestamp("TIME_MARK")))

    val eTime = dateUtils.getTimestamp(endDay)
    if (startDay != null && endDay != null) {
      val hoursArr = dateUtils.getHoursArr(startDay, endDay)
      for (i <- 0 until hoursArr.length - 1) {
        println("hoursArr.apply(i)："+hoursArr.apply(i))
        println("hoursArr.apply(i+1)："+hoursArr.apply(i+1))
        val hourRdd = rdd.filter(line => line._3.compareTo(hoursArr.apply(i)) >= 0 && line._3.compareTo(hoursArr.apply(i + 1)) < 0)
                         .map(x => x._2).collect()
        val hourEQ = hourRdd.apply(hourRdd.length - 1) - hourRdd.apply(0)

        println(hoursArr.apply(i) + " 到 " + hoursArr.apply(i + 1) + " 总电量为：" + hourEQ)
        hourEQArr += hourEQ
      }

    } else {
      
    }

    return hourEQArr
  }

  
  /**
   * 统计每小时的消耗电量
   */
  def getHourEQ(sql: String) {
    val rdd = new JdbcRDD(JdbcConnection.sc, JdbcConnection.getConnection, sql,
      1, 100, 1, r => (r.getString("NAME"), r.getDouble("VALUE"), r.getTimestamp("TIME_MARK")))
    val testRdd = rdd.groupBy(line => line._3.toString().subSequence(0, 13))
                     .collect().foreach(x => println("["+x._1.toString()+"]: "+(x._2.last._2 - x._2.head._2)))
    
    println("***********************共统计"+rdd.groupBy(line => line._3.toString().subSequence(0, 13)).count()+"个小时电量数据************************")
    
  }
  
  
  /**
   * 统计每天的电量
   */
  def getDayEQ(sql:String){
     val rdd = new JdbcRDD(JdbcConnection.sc, JdbcConnection.getConnection, sql,
      1, 100, 1, r => (r.getString("NAME"), r.getDouble("VALUE"), r.getTimestamp("TIME_MARK")))
    val testRdd = rdd.groupBy(line => line._3.toString().subSequence(0, 10))
                     .collect().foreach(x => println("["+x._1.toString()+"]: "+(x._2.last._2 - x._2.head._2)))
 
                     
    println("***********************共统计"+rdd.groupBy(line => line._3.toString().subSequence(0, 10)).count()+"天电量数据***************************")
  }
  
  
  /**
   * 统计每周的电量
   */
  def getWeekEQ(sql:String){
     val rdd = new JdbcRDD(JdbcConnection.sc, JdbcConnection.getConnection, sql,
      1, 100, 1, r => (r.getString("NAME"), r.getDouble("VALUE"), r.getTimestamp("TIME_MARK")))
    val testRdd = rdd.groupBy(line => dateUtils.getWeek(line._3))
                     .collect().foreach(x => println("["+x._1.toString()+"]: "+(x._2.last._2 - x._2.head._2)))
 
                     
    println("***********************共统计"+rdd.groupBy(line => dateUtils.getWeek(line._3)).count()+"个周电量数据***************************")
  }
  
  
  /**
   * 统计每月的电量
   */
  def getMonthEQ(sql:String){
     val rdd = new JdbcRDD(JdbcConnection.sc, JdbcConnection.getConnection, sql,
      1, 100, 1, r => (r.getString("NAME"), r.getDouble("VALUE"), r.getTimestamp("TIME_MARK")))
    val testRdd = rdd.groupBy(line => line._3.toString().subSequence(0, 7))
                     .collect().foreach(x => println("["+x._1.toString()+"]: "+(x._2.last._2 - x._2.head._2)))
 
                     
    println("***********************共统计"+rdd.groupBy(line => line._3.toString().subSequence(0, 7)).count()+"个月电量数据***************************")
  }
  
  object FastarUtils{
    def printHello(){
      print("Hello...")
    }
  }
}